package ex.other;

import ex.datastream.functions.sourceFunction.SourceFuncByItem;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class StreamingDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<SourceFuncByItem.Item>items=env.addSource(new SourceFuncByItem()).setParallelism(1);
        // 
//       SingleOutputStreamOperator<String>mapItems=items.map( new MyMapFunction());

        //銆恌latMap銆�

//        SingleOutputStreamOperator<Object>mapItems =items.flatMap(new FlatMapFunction<MyStreamingSource.Item, Object>() {
//            @Override
//            public void flatMap(MyStreamingSource.Item item, Collector<Object> collector) throws Exception {
//            String name=item.getName();
//            collector.collect(name);
//            }
//        });
//
        // 
//        SingleOutputStreamOperator<MyStreamingSource.Item>mapItems=items.filter(new FilterFunction<MyStreamingSource.Item>() {
//            @Override
//            public boolean filter(MyStreamingSource.Item item) throws Exception {
//                return item.getId()%2==0;
//            }
//        });
        // 
        SingleOutputStreamOperator<SourceFuncByItem.Item>mapItems=items.filter(item -> item.getId()%2==0);
        mapItems.print().setParallelism(1);
        String jobName="user defined streaming source";
        env.execute(jobName);

    }

    /**
     * 鑷畾涔塎ap
     */
    static class MyMapFunction extends RichMapFunction<SourceFuncByItem.Item,String> {

        @Override
        public String map(SourceFuncByItem.Item item) throws Exception {
            return item.getName();
        }
    }
}
